package com.navinfo.platform.thousandkilo.service;

import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.thousandkilo.dto.ThousandKiloData;
import com.navinfo.platform.thousandkilo.dto.TripInfoDto;
import com.navinfo.platform.thousandkilo.pojo.ThousandKiloPojo;
import com.navinfo.platform.thousandkilo.utils.HbaseConverter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

import static com.navinfo.platform.thousandkilo.service.ThousandKiloService.*;

public enum ThousandKiloDataHandleService implements IDataHandleService {
    // singleton
    INSTANCE;

    /**
     * 千公里区间 单位M
     */
    private static final int SECTION = 1000_000;

    private static Integer recognizeSection(Integer mileage) {
        if (mileage == null) {
            return 0;
        }
        if (mileage % SECTION == 0) {
            return mileage;
        }
        return (mileage / SECTION + 1) * SECTION;
    }

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        final Dataset<Row> tripInfo = castValue(obj.get(0));
        final Dataset<Row> maxSectionByTmn = castValue(obj.get(1));
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        final Broadcast<Map<String, String>> configMapBro = jsc.broadcast(configMap);
        //创建广播变量，广播每辆车目前最大区间
        JavaPairRDD<Long, ThousandKiloData> sectionPairRDD = maxSectionByTmn.toJavaRDD().mapPartitionsToPair(rowIterator -> {
                    List<Tuple2<Long, ThousandKiloData>> sectionList = new ArrayList<>();
                    while (rowIterator.hasNext()) {
                        Row row = rowIterator.next();
                        ThousandKiloData dto = new ThousandKiloData();
                        dto.setUid(Long.parseLong(row.getAs("uid").toString()));
                        dto.setTid(Long.parseLong(row.getAs("tid").toString()));
                        dto.setMileageSection(Integer.parseInt(row.getAs("mileageSection").toString()));
                        dto.setSectionStartFuel(Integer.parseInt(row.getAs("sectionStartFuel").toString()));
                        dto.setSectionEndFuel(Integer.parseInt(row.getAs("sectionEndFuel").toString()));
                        dto.setSectionFuel(Integer.parseInt(row.getAs("sectionFuel").toString()));
                        dto.setSectionStartMileage(Integer.parseInt(row.getAs("sectionStartMileage").toString()));
                        dto.setSectionEndMileage(Integer.parseInt(row.getAs("sectionEndMileage").toString()));
                        dto.setSectionMileage(Integer.parseInt(row.getAs("sectionMileage").toString()));
                        dto.setSectionUrea(Integer.parseInt(row.getAs("sectionUrea").toString()));
                        sectionList.add(new Tuple2<>(dto.getTid(), dto));
                    }
                    return sectionList.iterator();
                }
        );
        final Broadcast<Map<Long, ThousandKiloData>> sectionMapBroadcast = jsc.broadcast(new HashMap<>(sectionPairRDD.collectAsMap()));

        //转换tripinfo信息为一行
        Function<Row, TripInfoDto> tripInfoProvider = row -> {
            TripInfoDto tripInfoDto = new TripInfoDto();
            tripInfoDto.setTid(Long.parseLong(row.getAs("tid").toString()));
            tripInfoDto.setTripId(Integer.parseInt(row.getAs("trip_id").toString()));
            tripInfoDto.setRouteStartMileage(Integer.parseInt(Optional.ofNullable(row.getAs("route_start_mileage")).orElse(0).toString()));
            tripInfoDto.setRouteStartFuel(Integer.parseInt(Optional.ofNullable(row.getAs("route_start_fuel")).orElse(0).toString()));
            tripInfoDto.setRouteEndMileage(Integer.parseInt(Optional.ofNullable(row.getAs("route_end_mileage")).orElse(0).toString()));
            tripInfoDto.setRouteEndFuel(Integer.parseInt(Optional.ofNullable(row.getAs("route_end_fuel")).orElse(0).toString()));
            return tripInfoDto;
        };
        //转换tripinfo信息用list
        FlatMapFunction<Iterator<Row>, TripInfoDto> mapPartitionFunc = a -> {
            List<TripInfoDto> dl = new ArrayList<>();
            while (a.hasNext()) {
                Row row = a.next();
                if (row == null) {
                    continue;
                }
                TripInfoDto tripInfoDto = tripInfoProvider.call(row);
                dl.add(tripInfoDto);
            }
            return dl.iterator();
        };
        //转换为 <tid,List<tripinfo>>的键值对
        PairFlatMapFunction<Iterator<TripInfoDto>, Long, List<TripInfoDto>> pairPartitionFunction = tripInfoDtoPair -> {
            List<TripInfoDto> dataList = new ArrayList<>();
            tripInfoDtoPair.forEachRemaining(dataList::add);
            List<Tuple2<Long, List<TripInfoDto>>> newList = new ArrayList<>();
            newList.addAll(
                    dataList.stream().collect(Collectors.groupingBy(TripInfoDto::getTid))
                            .entrySet().stream().map(d -> new Tuple2<>(d.getKey(), d.getValue())).collect(Collectors.toList())
            );
            return newList.iterator();
        };
        //按照tid汇总
        Function2<List<TripInfoDto>, List<TripInfoDto>, List<TripInfoDto>> reduceFunc = (list1, list2) -> {
            list1.addAll(list2);
            return list1;
        };

        //根据区间数据组装最后保存的数据
        FlatMapFunction<Iterator<Tuple2<Long, List<TripInfoDto>>>, Tuple2<String, ThousandKiloData>> mapPartitionFunc2 = flagData -> {
            List<Tuple2<String, ThousandKiloData>> list = new ArrayList<>();
            while (flagData.hasNext()) {
                Tuple2<Long, List<TripInfoDto>> tuple2 = flagData.next();
                final Long tid = tuple2._1;

                List<TripInfoDto> tripInfoDtoList = tuple2._2();
                tripInfoDtoList.sort(Comparator.comparing(TripInfoDto::getTripId));

                Map<Long, ThousandKiloData> sectionMap = new HashMap<>(sectionMapBroadcast.getValue());
                // 更新或新增集合
                Map<String, ThousandKiloData> saveOrUpdateMap = initSaveOrUpdateMap(sectionMap.get(tid));

                Integer previousSection = null;
                for (TripInfoDto trip : tripInfoDtoList) {
                    final Integer routeEndMileage = trip.getRouteEndMileage();
                    final Integer targetSection = recognizeSection(routeEndMileage);
                    final Integer routeEndFuel = trip.getRouteEndFuel();
                    final Integer tripUrea = Optional.ofNullable(trip.getTripUrea()).orElse(0);
                    ThousandKiloPojo updateDt = new ThousandKiloPojo(routeEndMileage, routeEndFuel, tripUrea);

                    if (previousSection == null) {
                        // 第一条行程
                        ThousandKiloData queryData = findSection(sectionMap, tid, targetSection);
                        if (queryData == null) {
                            insertNewSection(saveOrUpdateMap, sectionMap, tid, targetSection, updateDt);
                        } else {
                            updateFuel(saveOrUpdateMap, queryData, updateDt);
                        }
                    } else {
                        // 非第一条行程
                        if (targetSection.equals(previousSection)) {
                            ThousandKiloData queryData = findSection(sectionMap, tid, targetSection);
                            if (queryData != null) {
                                updateFuel(saveOrUpdateMap, queryData, updateDt);
                            }
                        } else {
                            insertNewSection(saveOrUpdateMap, sectionMap, tid, targetSection, updateDt);
                        }
                    }
                    previousSection = targetSection;
                }
                list.addAll(saveOrUpdateMap.entrySet().stream().map(e -> new Tuple2<>(e.getKey(), e.getValue())).collect(Collectors.toList()));
            }
            return list.iterator();
        };

        VoidFunction<Iterator<Tuple2<String, ThousandKiloData>>> foreachFunc = s -> {
            Configuration configuration = HBaseConfiguration.create();
            configuration.set("hbase.zookeeper.quorum", configMap.get("hbase.zookeeper.quorum"));
            configuration.set("hbase.zookeeper.property.clientPort", configMap.get("hbase.zookeeper.property.clientPort"));
            configuration.set("zookeeper.znode.parent", configMap.get("zookeeper.znode.parent"));

            Connection conn = ConnectionFactory.createConnection(configuration);
            Table sectionTableName = conn.getTable(TableName.valueOf(configMapBro.getValue().get("save.thousandKilo.hbase.namespace") + ":"
                    + configMapBro.getValue().get("save.thousandKilo.hbase.tableName")));

            s.forEachRemaining(data -> {
                try {
                    sectionTableName.put(HbaseConverter.toPut(data._2()));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });

            sectionTableName.close();
            conn.close();
        };

        //1.将tripinfo信息转换为 <tid,List<tripinfo>>的键值对
        JavaPairRDD<Long, List<TripInfoDto>> pairRDD = tripInfo.toJavaRDD().mapPartitions(mapPartitionFunc)
                .mapPartitionsToPair(pairPartitionFunction)
                .reduceByKey(reduceFunc);
        //2.将聚合完的数据进行处理
        pairRDD.mapPartitions(mapPartitionFunc2)
                .foreachPartition(foreachFunc);
    }

    private void insertNewSection(Map<String, ThousandKiloData> saveOrUpdateMap, Map<Long, ThousandKiloData> sectionMap,
                                  Long tid, Integer section, ThousandKiloPojo updateDt) {
        // 更新前一个相邻区间
        if (section - SECTION > 0) {
            ThousandKiloData endSectionData = findSection(sectionMap, tid, section - SECTION);
            if (endSectionData != null) {
                updateFuel(saveOrUpdateMap, endSectionData, updateDt);
            }
        }
        // 追加新区间
        createData(saveOrUpdateMap, sectionMap, tid, section, updateDt);
    }
}
